package cdc.rxjava2.demo;

import cdc.tools.Log;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class Observer1 {
	//创建观察者或者订阅者
	Observer<String> observer = new Observer<String>() {
	    @Override
	    public void onSubscribe(Disposable d) {
	    //Disposable是1.x的Subscription改名的，因为Reactive-Streams规范用这个名称，为了避免重复
	    //这个回调方法是在2.0之后新添加的
	    //可以使用d.dispose()方法来取消订阅
	    }

	    @Override
	    public void onNext(String value) {
	        Log.e("onNext", value);
	    }

	    @Override
	    public void onError(Throwable e) {
	        Log.e("onError", e.getMessage());
	    }

	    @Override
	    public void onComplete() {
	        Log.e("onComplete", "complete");
	    }
	};
	
	//创建被观察者
	Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
	    @Override
	    public void subscribe(ObservableEmitter<String> e) throws Exception {
	    	e.onNext("Hello World!");
	    }
	});
	
	public Observer1() {
		observable.subscribe(observer);
	}
	
	

}
